package com.zynn.service.module.canal.listener;

import com.alibaba.otter.canal.protocol.CanalEntry;
import com.google.common.collect.Lists;
import com.google.common.collect.Maps;
import com.zynn.message.queue.constant.MqQueueNameConstant;
import com.zynn.message.queue.service.ScheduleMessagePostProcessor;
import com.zynn.message.queue.service.SenderService;
import com.zynn.service.module.canal.constant.TableNameConstant;
import com.zynn.service.module.canal.enums.ProfileMnitorColumn;
import com.zynn.service.module.canal.event.UpdateCanalEvent;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.collections.MapUtils;
import org.apache.commons.lang.StringUtils;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Component;

import java.util.*;


/**
 * @author jerry-liqi
 *
 * 监听修改
 */
@Component
@Slf4j
public class UpdateCanalListener extends AbstractCanalListener<UpdateCanalEvent> {

    @Autowired
    private SenderService senderService;


    /**
     *  同步执行
     *  @param database 数据库
     *  @param table 表名
     *  @param rowData 数据
     */
    @Override
    protected void doSync(String database, String table, CanalEntry.RowData rowData) {
        log.info("UpdateCanalListener："+ database + table);

        Map<String, Object> columnsMap = super.parseColumnsToMap(rowData.getAfterColumnsList());

        /**
         * 推荐V3 推荐可能认识人
         */
        this.recommendFriendMqQueueV3Monitor(table,columnsMap);


        /**
         * 1.数据迁移
         * @deprecated
         * this.migrateData(table,columnsMap);
         */

        /**
         * 2.同步es用户信息
         * @deprecated
         * this.synEsUsers(table,columnsMap);
         */

        /**
         * 3.个人资料修改时包括生日，城市，职业，学校，学院，年级生成首页动态
         * this.createEventsForModifyProfileInfo(table,rowData);
         */
    }

    /**
     * 数据迁移
     * @param table
     * @param columnsMap
     */
    private void migrateData(String table,Map<String, Object> columnsMap){
        if (TableNameConstant.GROUPS.equals(table)) {
            senderService.send(MqQueueNameConstant.MIGRATE_GROUP_ON_TIME, columnsMap);
        }
        if (TableNameConstant.GROUPMEMBERS.equals(table)) {
            senderService.send(MqQueueNameConstant.MIGRATE_GROUP_MEMBERS_ON_TIME, columnsMap);
        }
    }

    /**
     * 同步es用户信息
     * @param table 表名
     * @param columnsMap 表元数据
     */
    private void synEsUsers(String table,Map<String, Object> columnsMap){
        if (TableNameConstant.USERS.equals(table)) {
            //同步es
            senderService.send(MqQueueNameConstant.SYN_ES_USERS, columnsMap);
        }
    }


    /**
     * 推荐V3 用户学校信息监控
     * @param table 表名
     * @param columnsMap 表元数据
     */
    private void recommendFriendMqQueueV3Monitor(String table,Map<String, Object> columnsMap){
        if (TableNameConstant.BASE_USER_INFO.equals(table)) {
            senderService.send(MqQueueNameConstant.FIND_FRIEND_FORM_BASE_USER_INFO, columnsMap,new ScheduleMessagePostProcessor(1000L,0L,null,null));
        }
    }

    /**
     * 个人资料修改时包括生日，城市，职业，学校，学院，年级生成首页动态
     * @param table
     * @param rowData
     */
    private void createEventsForModifyProfileInfo(String table,CanalEntry.RowData rowData){
        /**base_user_info**/
        if(TableNameConstant.BASE_USER_INFO.equals(table)) {
            Map<String, Object> columnsMapBefore = super.parseColumnsToMap(rowData.getBeforeColumnsList());
            Map<String, Object> columnsMapAfter = super.parseColumnsToMap(rowData.getAfterColumnsList());
            Map<String,Object> eventColumnMap = generateResultMap(columnsMapBefore, columnsMapAfter, Arrays.asList(ProfileMnitorColumn.BASE_USER_INFO_COLUMNS.getColumns()));
            if(MapUtils.isNotEmpty(eventColumnMap)){
                /**添加用户字段**/
                eventColumnMap.put("user_id",columnsMapAfter.get("user_id"));
                senderService.send(MqQueueNameConstant.PROFILE_MODIFIED_CREATE_EVENT,eventColumnMap);
            }
        }

        /**users**/
        if(TableNameConstant.USERS.equals(table)) {
            Map<String, Object> columnsMapBefore = super.parseColumnsToMap(rowData.getBeforeColumnsList());
            Map<String, Object> columnsMapAfter = super.parseColumnsToMap(rowData.getAfterColumnsList());
            Map<String,Object> eventColumnMap = generateResultMap(columnsMapBefore, columnsMapAfter, Arrays.asList(ProfileMnitorColumn.USERS_COLUMNS.getColumns()));
            if(MapUtils.isNotEmpty(eventColumnMap)){
                /**添加用户字段**/
                eventColumnMap.put("user_id",columnsMapAfter.get("userid"));
                senderService.send(MqQueueNameConstant.PROFILE_MODIFIED_CREATE_EVENT,eventColumnMap);
            }
        }
    }

    /**
     * @param columnsMapBefore
     * @param columnsMapAfter
     * @param columns
     * @return
     */
    private Map<String,Object> generateResultMap(Map<String, Object> columnsMapBefore, Map<String, Object> columnsMapAfter, List<String> columns) {
        Map<String, Object> eventColumnMap = Maps.newHashMap();
        Object before;
        Object after;
        for (String column: Optional.ofNullable(columns).orElse(Lists.newArrayList())) {
            before = columnsMapBefore.get(column);
            after = columnsMapAfter.get(column);
            if(!Objects.equals(before,after)){
                if(Objects.nonNull(after)){
                    if(StringUtils.equals(column,"ucity")){
                        after = String.valueOf(columnsMapAfter.get("uprovince"))+"省"+after + "市";
                    }
                    eventColumnMap.put(column,after);
                }
            }
        }
        return eventColumnMap;
    }
}
